package app.socket.client;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import app.htby.utils.KafkaConsumeUtils;
import app.htby.utils.SocketUtils;
import app.htby.utils.SpringContextUtil;
import app.kafka.consume.DisWebConsumer;
import app.kafka.consume.DisWebConsumerRunnable;

/**
 * 分发系统的 socket异步连接
 * 
 * @author lfy.xys
 * @date 2018年5月22日
 *
 */
public class DisConnClientRunnable implements Runnable {

	private static Logger logger = LoggerFactory.getLogger(DisConnClientRunnable.class);

	private String ip;
	private int port;
	private byte[] sendMsg;// 发送的消息
	private ByteBuffer rBuffer = ByteBuffer.allocate(1024 * 9);
	private int count = 0;

	public DisConnClientRunnable(String ip, int port, byte[] sendMsg) {
		this.ip = ip;
		this.port = port;
		this.sendMsg = sendMsg;
	}

	@Override
	public void run() {
		try {
			InetSocketAddress SERVER = new InetSocketAddress(ip, port);
			SocketChannel socketChannel = SocketChannel.open();
			socketChannel.configureBlocking(false);
			Selector selector = Selector.open();
			socketChannel.register(selector, SelectionKey.OP_CONNECT);
			socketChannel.connect(SERVER);
			SocketChannel client = null;

			while (true) {
				selector.select();
				Set<SelectionKey> keySet = selector.selectedKeys();
				for (final SelectionKey selectionKey : keySet) {
					if (selectionKey.isConnectable()) {
						client = (SocketChannel) selectionKey.channel();
						if (client.isConnectionPending()) {
							try {
								client.finishConnect();
							} catch (Exception e) {
								reconnect();
							}
							logger.info("connect success ! sendMsg : {}", new String(sendMsg));
							// 发送数据
							client.write(ByteBuffer.wrap(sendMsg));
							// 修改 sendMsg 为 断线重连状态
							sendMsg = SocketUtils.DIS_RECONNECT_MANAGE.getBytes();
						}
						client.register(selector, SelectionKey.OP_READ);
					} else if (selectionKey.isReadable()) {
						boolean failure = true;
						client = (SocketChannel) selectionKey.channel();
						rBuffer.clear();
						try {
							count = client.read(rBuffer);
							// logger.info("count:" + count);
							failure = false;
						} catch (Exception e) {
							logger.info("断开连接");
							e.printStackTrace();
						}

						if (count <= 0 || failure) {
							client.close();
							selectionKey.cancel(); // Some JDK implementations run into an infinite loop without this.
							reconnect();
						} else {
							// 重新读取 指定长度的数据。 否则获取的是创建的缓冲区长度，末尾会被补0。
							rBuffer.clear();
							byte[] bytes = new byte[count];
							rBuffer.get(bytes, 0, bytes.length);
							client.register(selector, SelectionKey.OP_READ);

							String msgStr = new String(bytes);//客户端收到的信息
							logger.info("客户端收到的信息：{}" ,msgStr);
							CreateKafka(msgStr);// 创建消费者
						}
					}
				}
				keySet.clear();
			}

		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 服务端连接不上，客户端断线重连
	 * 
	 * @author lfy.xys
	 * @date 2018年5月30日
	 *
	 */
	public void reconnect() {
		logger.info("服务端连接失败，{}ms后进行重连...", SocketUtils.NIO_RECONNECT_INTERVAL_TIME);
		try {
			Thread.sleep(SocketUtils.NIO_RECONNECT_INTERVAL_TIME);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		this.run();
	}

	public byte[] getSendMsg() {
		return sendMsg;
	}

	public void setSendMsg(byte[] sendMsg) {
		this.sendMsg = sendMsg;
	}

	public void CreateKafka(String msgStr) {
		//根据 && 切分
		String[] msgs = msgStr.split("&&");
		logger.info("消费的topic：{}",msgs.toString());
		for (String topic : msgs) {
			
			logger.info("消费的topic：{}",topic);
			//判断 每个 topic 是否已经被 消费. 存在则被消费过
			if(!KafkaConsumeUtils.topicSet.contains(topic)){
				// 没有被消费过的topic
				String beanId = "Consumer_" + KafkaConsumeUtils.topicCount;//设置id
				SpringContextUtil.registerBean(beanId, DisWebConsumer.class.getName());// 使用spring创建bean
				// 创建线程消费
				new Thread(new DisWebConsumerRunnable(topic,beanId)).start();
			}
		}
		
	}
	
	public static void main(String[] args) {
		DisConnClientRunnable r = new DisConnClientRunnable("localhost", 8888, SocketUtils.DIS_START_STATE.getBytes());
		r.run();
	}

}
